package org.example.flinktest.apidemo;

import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.example.flinktest.bean.SensorReading;
import org.example.flinktest.operators.MakeSensorMapFunction;

/**
 * @author shihb
 * @date 2019/12/11 18:27
 */
public class WindowDemo {

  public static void main(String[] args) throws Exception {

    //环境
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(1);
    //设置时间触发
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

    //source

    DataStream<String> source =env.socketTextStream("10.0.75.2",7777);


    //transform
    DataStream<SensorReading> dataStream=source
        .map(new MakeSensorMapFunction())
        .assignTimestampsAndWatermarks(
        new BoundedOutOfOrdernessTimestampExtractor<SensorReading>(Time.seconds(1)) {
          @Override
          public long extractTimestamp(SensorReading element) {
            return element.getTimestamp()*1000;
          }
        });

    //统计10秒内的最小温度
    DataStream<Tuple2<String, Double>> minTempStream = dataStream
        .keyBy("id")
        .timeWindow(Time.seconds(10),Time.seconds(5))
        .aggregate(
            new AggregateFunction<SensorReading, Tuple2<String, Double>, Tuple2<String, Double>>() {
              @Override
              public Tuple2<String, Double> createAccumulator() {
                return new Tuple2<>("acc", Double.MAX_VALUE);
              }

              @Override
              public Tuple2<String, Double> add(SensorReading sensorReading,
                  Tuple2<String, Double> acc) {
                return new Tuple2<>(sensorReading.getId(),
                    Math.min(sensorReading.getTemperature(), acc.f1));
              }

              @Override
              public Tuple2<String, Double> getResult(Tuple2<String, Double> acc) {
                return acc;
              }

              @Override
              public Tuple2<String, Double> merge(Tuple2<String, Double> acc1,
                  Tuple2<String, Double> acc2) {
                return new Tuple2<>(acc1.f0,Math.min(acc1.f1,acc2.f1));
              }
            });
    //sink
    minTempStream.print("min temp data");
    dataStream.print("input data");





    env.execute("window job test");

  }

}
